package com.linkstec.kafka;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

public class PushLagMonitor {
    private final static String TOPIC = "test";
    private final static String BOOTSTRAP_SERVERS = "192.168.2.232:9092,192.168.2.233:9092,192.168.2.234:9092";
    private final static String ZOOKEEPER = "192.168.2.232:2181,192.168.2.233:2181,192.168.2.234:2181";
    private static Consumer<Long, String> consumer = null;
    private static BaseZookeeper zookeeper = new BaseZookeeper();

    static  {
        final Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaExampleConsumer");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        consumer = new KafkaConsumer<>(props);
        zookeeper = new BaseZookeeper();
        try {
			zookeeper.connectZookeeper(ZOOKEEPER);
		} catch (Exception e) {
			e.printStackTrace();
		}
    }

    public static void getPartitionsForTopic() {

        Collection<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC);
        System.out.println("Get the partition info as below:");
        List<TopicPartition> tp =new ArrayList<TopicPartition>();
        partitionInfos.forEach(str -> {
            System.out.println("Partition Info:");
            System.out.println(str);

            tp.add(new TopicPartition(TOPIC,str.partition()));
            consumer.assign(tp);
            consumer.seekToEnd(tp);

            System.out.println("Partition " + str.partition() + " 's latest offset is '" + consumer.position(new TopicPartition(TOPIC, str.partition())));
        });
    }

    public static void main(String[] args) {
		getPartitionsForTopic();
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

        try {
			List<String> children = zookeeper.getChildren("/consumers/kafka-dbsycn-service/offsets/test");
			System.out.println(children);
			for (String partition : children) {
				String offset = zookeeper.getData("/consumers/kafka-dbsycn-service/offsets/test/" + partition);
				System.out.println("Partition " + partition + "的offset是：" + offset);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}
}